package com.tx.mq.consumer;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeoutException;

public class DlConsumerTest {

    private static final String DEFAULT_DL_EXCHANGE_PROP = "x-dead-letter-exchange";
    private static final String DEFAULT_DL_QUEUE = "DL-QUEUE";
    private static final String DEFAULT_DL_TEST_QUEUE = "DL-TEST-QUEUE";

    public static void main(String[] args) throws IOException, TimeoutException {
        ConnectionFactory cf = new ConnectionFactory();
        cf.setHost("localhost");

        Connection connection = cf.newConnection();
        Channel channel = connection.createChannel();

        String testDlExchange = "test-dl-exchange";
        String dlExchange = "dl-exchange";
        String dlRouterKey = "dl.test";

        // 测试队列设置
        Map<String, Object> map = new HashMap<>();
        map.put(DEFAULT_DL_EXCHANGE_PROP, dlExchange);
        channel.exchangeDeclare(testDlExchange, BuiltinExchangeType.TOPIC, true, false, null);
        channel.queueDeclare(DEFAULT_DL_TEST_QUEUE, true, false, false, map);
        channel.queueBind(DEFAULT_DL_TEST_QUEUE, testDlExchange, dlRouterKey);

        // 死信队列设置
        channel.exchangeDeclare(dlExchange, BuiltinExchangeType.TOPIC, true, false, null);
        channel.queueDeclare(DEFAULT_DL_QUEUE, true, false, false, null);
        channel.queueBind(DEFAULT_DL_QUEUE, dlExchange, "#");

        DeliverCallback callback = (consumerTag, delivery) -> {
            System.out.println(consumerTag + " " + new String(delivery.getBody()));
            // 此处如果requeue设置为true，在消息的ttl时间内不断重试，超时则进入死信队列
            channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false, true);
        };

        // 注释之后可以查看消息超时进入死信队列，否则回调直接进入私信队列看不到时间差
        channel.basicConsume(DEFAULT_DL_TEST_QUEUE, false, callback, (consumerTag -> {}));
    }
}
